package com.yukefms.engine;

import com.yukefms.engine.interfaces.PointCloudReader;
import com.yukefms.jointpoint.BodyPointCloud;
import com.yukefms.message.MessageContainer;
import com.yukefms.message.MessageTools;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.stereotype.Component;

/**
 * @Description
 * The class is defined for pointCloud reading, which relies on a message container to cache all readed/produced data.
 * For caching readed data, the message container needs a message consumer which is based on KafkaConsumer to read
 * data from Kafka.( PointCloudReader --d--> MessageContainer --d--> MessageConsumer --d--> KafkaConsumer )
 *
 * @Author Ping
 * @Date 2021/6/26 10:39
 **/
@Component
public class DefaultPointCloudReader implements PointCloudReader {
    public static Logger logger = LogManager.getLogger(DefaultPointCloudReader.class);
    private MessageContainer<String> container ;

    public DefaultPointCloudReader() {
        this.container = new MessageContainer<>();
    }

    @Override
    public boolean ended() {
        return this.container.ended();
    }

    @Override
    public void stop() {
        this.container.stopWork();
    }
    @Override
    public void start() {
        this.container.startWork() ;
    }

    @Override
    public BodyPointCloud next() {
        // NOTE THAT, here should considerring some exceptional situations,
        // e.g., data is empty for some while.
        BodyPointCloud pc = null ;
       // this.container.pop() ;
        if (this.container.isStarted()) {
            if (this.container.empty()) return null ;
            pc = MessageTools.toPointCloud(this.container.pop()) ;
        }
        return pc;
    }
}
